package com.my.screenadapter.rxjavashow;

import android.graphics.Bitmap;
import android.graphics.drawable.Drawable;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.ImageView;
import android.widget.LinearLayout;
import android.widget.Toast;

import com.my.screenadapter.Base.BaseActivity;
import com.my.screenadapter.utils.MyViewUtils;
import com.my.screenadapter.R;
import com.my.screenadapter.rxjavashow.guanchazhe.SubscriptionSubject;
import com.my.screenadapter.rxjavashow.guanchazhe.WeiXinUser;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import butterknife.BindView;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by 张英杰 on 2017/3/1.
 * <p>
 * 文章：
 * 给 Android 开发者的 RxJava1.0详解
 * http://gank.io/post/560e15be2dca930e00da1083
 * 给初学者的RxJava2.0教程(一)
 * http://www.jianshu.com/p/464fa025229e
 * 观察者模式模式简介
 * http://blog.csdn.net/itachi85/article/details/50773358
 */
public class RxJavaActivity extends BaseActivity {
    private final String TAG = "RxJavaActivity===";
    @BindView(R.id.ll)
    LinearLayout mllcontainer;
    @BindView(R.id.iv)
    ImageView iv;
    @BindView(R.id.btn1)
    Button btn1;
    @BindView(R.id.btn2)
    Button btn2;
    //注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.
    View.OnClickListener mRxjava2Disposable = new View.OnClickListener() {
        @Override
        public void onClick(View v) {
            Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    Log.d(TAG, "emit 1");
                    emitter.onNext(1);
                    Log.d(TAG, "emit 2");
                    emitter.onNext(2);
                    Log.d(TAG, "emit 3");
                    emitter.onNext(3);
                    Log.d(TAG, "emit complete");
                    emitter.onComplete();
                    Log.d(TAG, "emit 4");
                    emitter.onNext(4);
                }
            }).subscribe(new Observer<Integer>() {
                private Disposable mDisposable;
                private int i;

                @Override
                public void onSubscribe(Disposable disposable) {
                    Log.d(TAG, "subscribe");
                    mDisposable = disposable;
                }

                @Override
                public void onNext(Integer integer) {
                    Log.d(TAG, "ObserveronNext: " + integer);
                    i++;
                    if (i == 2) {
                        Log.d(TAG, "dispose");
                        mDisposable.dispose();
                        Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
                    }
                }

                @Override
                public void onError(Throwable throwable) {
                    Log.d(TAG, "error");
                }

                @Override
                public void onComplete() {
                    Log.d(TAG, "complete");
                }
            });
        }
    };
    /**
     * 最为关键的是onComplete和onError必须唯一并且互斥,
     * 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
     */
    View.OnClickListener mRxjava2Demo = new View.OnClickListener() {
        @Override
        public void onClick(View v) {
            //创建一个上游 Observable：
            Observable<Integer> observable = Observable
                    .create(new ObservableOnSubscribe<Integer>() {
                        //                        ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，
//                        它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)
//                        就可以分别发出next事件、complete事件和error事件。
                        @Override
                        public void subscribe(ObservableEmitter<Integer> observableEmitter) throws Exception {
                            observableEmitter.onNext(1);
                            observableEmitter.onNext(2);
                            observableEmitter.onNext(3);
                            observableEmitter.onComplete();
                        }
                    });
            //创建一个下游 Observer
            Observer<Integer> observer = new Observer<Integer>() {
                @Override
                public void onSubscribe(Disposable disposable) {
                    Log.d(TAG, "subscribe");
                }

                @Override
                public void onNext(Integer value) {
                    Log.d(TAG, "--observer接收--" + value);
                }

                @Override
                public void onError(Throwable throwable) {
                    Log.d(TAG, "error");
                }

                @Override
                public void onComplete() {
                    Log.d(TAG, "complete");
                }
            };
            //建立连接
            observable.subscribe(observer);
        }
    };
    //mRxjava2Demo
    //事件变换
    View.OnClickListener mChangeListener = new View.OnClickListener() {
        @Override
        public void onClick(View v) {
            Observable.just("images/logo.png") // 输入类型 String
                    .map(new Function<String, Bitmap>() {
                        @Override
                        public Bitmap apply(@NonNull String s) throws Exception { // 参数类型 String
                            return null;// 返回类型 Bitmap
                        }
                    }).subscribe();
        }
    };
    /**
     * 观察者模式（又被称为发布-订阅（Publish/Subscribe）模式，
     * 属于行为型模式的一种，它定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。
     * 这个主题对象在状态变化时，会通知所有的观察者对象，使他们能够自动更新自己。
     * <p>
     * Subject：抽象主题（抽象被观察者），
     * 抽象主题角色把所有观察者对象保存在一个集合里，每个主题都可以有任意数量的观察者，
     * 抽象主题提供一个接口，可以增加和删除观察者对象。
     * ConcreteSubject：具体主题（具体被观察者），
     * 该角色将有关状态存入具体观察者对象，在具体主题的内部状态发生改变时，
     * 给所有注册过的观察者发送通知。
     * Observer：抽象观察者，是观察者者的抽象类，它定义了一个更新接口，
     * 使得在得到主题更改通知时更新自己。
     * ConcrereObserver：具体观察者，是实现抽象观察者定义的更新接口，
     * 以便在得到主题更改通知时更新自身的状态。
     */
    View.OnClickListener mGuanChaZhe = new View.OnClickListener() {
        @Override
        public void onClick(View v) {
            //客户端调用
            SubscriptionSubject mSubscriptionSubject = new SubscriptionSubject();
            //创建微信用户
            WeiXinUser user1 = new WeiXinUser(RxJavaActivity.this, "杨影枫");
            WeiXinUser user2 = new WeiXinUser(RxJavaActivity.this, "月眉儿");
            WeiXinUser user3 = new WeiXinUser(RxJavaActivity.this, "紫轩");
            //订阅公众号
            mSubscriptionSubject.attach(user1);
            mSubscriptionSubject.attach(user2);
            mSubscriptionSubject.attach(user3);
            //公众号更新发出消息给订阅的微信用户
            mSubscriptionSubject.notify("刘望舒的专栏更新了");
        }
    };

    @Override
    protected void init() {
        btn1.setText("出发Rxjava 显示Drawable");
        btn1.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                helloRxJavaDrawable();
            }
        });
        btn2.setText("Scheduler显示log 线程控制");
        btn2.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        for (int i = 0; i < 3; i++) {
                            helloScheduler(i);
                        }
                    }
                }).start();

            }
        });

        MyViewUtils.addButton(mllcontainer, new Button(this), "观察者模式Demo", mGuanChaZhe);
        MyViewUtils.addButton(mllcontainer, new Button(this), "事件变换", mChangeListener);
        MyViewUtils.addButton(mllcontainer, new Button(this), "Rxjava2.0————Demo", mRxjava2Demo);
        MyViewUtils.addButton(mllcontainer, new Button(this), "Rxjava2.0————Disposable", mRxjava2Disposable);

    }

    private static final Lock lock = new ReentrantLock();
    /**
     * 在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
     * 1) Scheduler 的 API (一)
     * 在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ，它们已经适合大多数的使用场景：
     * Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
     * Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
     * Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
     * Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
     * 另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
     * 有了这几个 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。 * subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。 * observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
     */
    private synchronized void helloScheduler(final Integer integer) {
//        lock.lock();
//        Observable.just(1, 2, 3, 4)
        Log.i(TAG, "11---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> observableEmitter) throws Exception {
                Log.i(TAG, "22---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
                observableEmitter.onNext(integer);
                observableEmitter.onComplete();
                Thread.sleep(1000);
            }
        })
                .subscribeOn(Schedulers.single()) //可同步执行
                .observeOn(Schedulers.single())
//                .observeOn(Schedulers.trampoline())  //可同步执行
//                .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
//                .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
//                .delay(3, TimeUnit.SECONDS)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "23---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "33---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName()+"   data=="+integer);
//                        Toast.makeText(RxJavaActivity.this,"接收到"+integer,Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "44---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
//                        Log.i(TAG, "55---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
//                        lock.unlock();
                    }
                });
    }

    //显示测试显示图片
    private void helloRxJavaDrawable() {
        final int drawableRes = R.drawable.xiaohei;
        Log.i(TAG, "11---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
        Observable.create(new ObservableOnSubscribe<Drawable>() {
            @Override
            public void subscribe(ObservableEmitter<Drawable> observableEmitter) throws Exception {
                //    Drawable drawable = getTheme().getDrawable(drawableRes);//sdk最小21
                Drawable drawable = getResources().getDrawable(drawableRes);
                observableEmitter.onNext(drawable);
                observableEmitter.onComplete();
                Log.i(TAG, "22---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Drawable>() {
                    @Override
                    public void onSubscribe(Disposable disposable) {

                    }

                    @Override
                    public void onNext(Drawable drawable) {
                        iv.setImageDrawable(drawable);
                        Log.i(TAG, "33---ThreadId==" + Thread.currentThread().getId() + "  name==" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        Toast.makeText(RxJavaActivity.this, "Error!", Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //观察者模式模式简介Demo

    @Override
    protected int initLayoutId() {
        return R.layout.activity_rxjava;
    }
}
